package com.thh.zookeeperlock.lock;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class ZookeeperLock implements Lock {

    // 锁的工作区间,用来隔离和区分
    private static final String WORKSPACE = "/lock-workspace";

    // 锁的名称
    private String lockName;

    //第三方的zookeeper客户端
    private CuratorFramework client;

    // Zookeepeer集群节点名
    //public static final String ZOOKEEPERSTRING = "192.168.99.129:2181,192.168.99.153:2181,192.168.99.171:2181";
    public static final String ZOOKEEPERSTRING = "127.0.0.1:2181";

    //为什么传递lockName,因为开发的时候不仅仅是一种业务需要加锁，通用性应该让用户自己去决定不同业务锁的名字。
    public ZookeeperLock(String lockName){
        this.lockName = lockName;
        init();
    }

    // 初始化
    private void init(){
        RetryPolicy policy = new ExponentialBackoffRetry(1000, 10);
        client = CuratorFrameworkFactory.newClient(ZOOKEEPERSTRING,5000,1000,policy);
        client.start();
        try{
            //创建工作区间
            if(client.checkExists().forPath(WORKSPACE)==null){
                //如果需要就创建父节点
                client.create().creatingParentsIfNeeded()
                        // 创建节点类型，持久节点
                        .withMode(CreateMode.PERSISTENT)
                        // 权限类型，暂时不设置
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        // 创建目录
                        .forPath(WORKSPACE);
            }
        }catch(Exception ex){
            ex.printStackTrace();
        }
    }


    @Override
    public void lock() {
        // 自旋，不停的获取监听别的请求。锁重入
        while (true){
            String lockPath = WORKSPACE + "/" + lockName;
            try{
                //判断锁节点是否存在，如果不存在就创建
                if(client.checkExists().forPath(lockPath)==null){
                    //如果需要就创建父节点
                    client.create().creatingParentsIfNeeded()
                            // 创建节点类型，持久节点
                            .withMode(CreateMode.EPHEMERAL)
                            // 权限类型，暂时不设置
                            .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                            // 创建目录
                            .forPath(lockPath);
                    System.out.println(lockName + "======================>获取到Lock了....");
                    return;
                }else{
                    // 监听进入阻塞状态
                    listenerWait();
                }
            }catch(Exception ex){ // 为什么异常也要监听，是因为如果节点已经创建成功，另外的进程进入来的时候，会抛出异常ZNodeExistException异常，这个时候说明lock锁还没释放。
                try{
                    // 监听进入阻塞状态
                    listenerWait();
                }catch(Exception exe){
                    //exe.printStackTrace();
                }
            }
        }
    }

    public void listenerWait() throws  Exception{
        // 并发编程类
        final  CountDownLatch countDownLatch = new CountDownLatch(1);

        // 初始化子节点监听器
        PathChildrenCache childrenCache = new PathChildrenCache(client,WORKSPACE,true);
        childrenCache.start();

        //添加监听事件
        childrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override//
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                //判断监听的事件类型,监听孩子节点是否删除，删除以后就释放锁
                if(event.getData().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)
                 && event.getData().getPath().contains(lockName)){
                    System.out.println("子节点删除了，开始释放锁 " + lockName);
                    //唤醒其他阻塞的线程
                    countDownLatch.countDown();
                }
            }
        });

        //阻塞进程，不往下执行
        countDownLatch.wait();
    }


    @Override
    public void unlock() {
        String lockPath = WORKSPACE + "/" + lockName;
        try{
            //判断锁节点是否存在，如果不存在就创建
            if(client.checkExists().forPath(lockPath)!=null) {
                client.delete().forPath(lockPath);
                System.out.println(lockName + "======================>锁释放了");
            }
        }catch(Exception ex){
            ex.printStackTrace();
        }
    }


    @Override
    public boolean tryLock() {
        return false;
    }


    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }


    @Override
    public Condition newCondition() {
        return null;
    }
}
